package com.heima.schedule.service.Impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.constants.ScheduleConstants;
import com.heima.common.redis.CacheService;
import com.heima.model.schedule.dtos.TaskDto;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.List;
import java.util.Set;

@Service
@Slf4j
@Transactional
public class TaskServiceImpl implements TaskService {

    @Autowired
    private TaskinfoMapper taskinfoMapper;
    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;
    @Autowired
    private CacheService cacheService;

    /**
     * 添加任务
     *
     * @param dto 任务对象
     * @return 任务Id
     */
    @Override
    public long addTask(TaskDto dto) {
        //给dto设置 id（雪花算法）
        dto.setTaskId(IdWorker.getId());

        //将任务和任务日志添加到数据库
        boolean result = addTaskAndLogToDb(dto);

        if (result) {
            //保存数据库成功
            //将符合条件的任务添加到对应的Redis中
            addTaskToRedis(dto);
        }
        return dto.getTaskId();
    }

    /**
     * 1: 当前任务时间 <= 当前时间，存放入 list中
     * 2： 当前任务时间 > 当前时间 ，<= 预设时间（当前时间 + 5分钟）  存放如zset中
     *
     * @param dto
     */
    private void addTaskToRedis(TaskDto dto) {
        //预设时间：当前时间 + 5 分钟 （1000 * 60 *5）
        long zSetTime = System.currentTimeMillis() + (1000 * 60 * 5);

        //设置Key：定时任务的 类型 和 优先级
        String key = dto.getTaskType() + "_" + dto.getPriority();
        //1: 当前任务时间 <= 当前时间，存放入 list中  lLeftPush
        if (dto.getExecuteTime() <= System.currentTimeMillis()) {
            cacheService.lLeftPush(ScheduleConstants.TOPIC + key,
                    JSON.toJSONString(dto));
        } else if (dto.getExecuteTime() > System.currentTimeMillis() && dto.getExecuteTime() <= zSetTime) {
            //当前任务时间 > 当前时间 ，<= 预设时间（当前时间 + 5分钟）  存放如zset中  zAdd
            cacheService.zAdd(ScheduleConstants.FUTURE + key,
                    JSON.toJSONString(dto),
                    dto.getExecuteTime());
        }
    }

    /**
     * 添加任务对象和任务日志达到数据库
     *
     * @param dto
     * @return
     */
    private boolean addTaskAndLogToDb(TaskDto dto) {
        boolean flag = false;

        try {
            //保存任务表
            Taskinfo taskinfo = new Taskinfo();
            BeanUtils.copyProperties(dto, taskinfo);
            taskinfo.setExecuteTime(new Date(dto.getExecuteTime()));
            //保存到数据库
            taskinfoMapper.insert(taskinfo);

            //设置TaskDto 外面设置了
            //dto.setTaskId(taskinfo.getTaskId());

            //保存任务日志数据
            TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
            BeanUtils.copyProperties(taskinfo, taskinfoLogs);
            taskinfoLogs.setVersion(1);//版本号设置默认为 1
            taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);//状态设置为 初始化
            //保存到数据库
            taskinfoLogsMapper.insert(taskinfoLogs);
            log.info("定时任务数据成功保存到数据库中...");
            flag = true;
        } catch (Exception e) {
            e.printStackTrace();
            log.info("定时任务数据保存数据库失败。");
        }

        return flag;
    }

    /**
     * 拉取消费任务
     * 从List右边获取一个TaskDto
     * 将Taskinfo中该条数据删除
     * 修改Log表中数据信息
     *
     * @param type
     * @param priority
     * @return
     */
    @Override
    public TaskDto popTak(Integer type, Integer priority) {
        TaskDto dto = null;
        try {
            //从Redis 的 List中取出 并删除
            //拼接Key
            String key = ScheduleConstants.TOPIC + type + "_" + priority;
            //从Redis中取出并删除  lRightPop（从右边拿出来一个）
            String taskJson = cacheService.lRightPop(key);

            if (StringUtils.isNotEmpty(taskJson)) {
                //取出的数据转为 TaskDto对象
                dto = JSON.parseObject(taskJson, TaskDto.class);

                //更新数据库中的数据
                //将Taskinfo 表中数据删除
                taskinfoMapper.deleteById(dto.getTaskId());

                //将TaskinfoLogs 表中数据状态更新
                TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(dto.getTaskId());
                taskinfoLogs.setStatus(ScheduleConstants.EXECUTED);
                taskinfoLogsMapper.updateById(taskinfoLogs);
                log.info("拉取任务成功，数据库已修改。");
            } else {
                log.info("List中无延时任务数据");
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.info("拉取任务失败");
        }
        return dto;
    }


    /**
     * redis 定时刷新 每分钟一次
     * 将zset  中 符合条件 的数据 传到 list中 ，并且将zset该数据 删除
     */
    @Scheduled(cron = "0 */1 * * * ? ")//定时任务：每分钟执行一次
    public void refrshRedis() {

        // Redis分布式锁  上锁： //结果返回：token，锁的内容，如果不为空则：请求锁成功
        String token = cacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
        if (StringUtils.isNotEmpty(token)) {
            log.info("数据转移定时任务开始...");

            //查询Redis中的建 ： scan模糊查询全部
            //查询到所有以 future_开头的Key
            Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");

            //遍历：山选出符合条件的数据 ： 分数 <= 当前时间的毫秒值
            for (String futureKey : futureKeys) {

                //筛选出出符合条件的数据 ： 使用工具类：zRangeByScore 筛选出分数小于当前时间的 value
                Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());

                //构建List 的key
                String listKey = ScheduleConstants.TOPIC + futureKey.split(ScheduleConstants.FUTURE)[1];

                //参数校验：如果 符合条件的数据不为空 则开始 转移
                if (tasks != null && tasks.size() > 0) {
                    //存放进 List 中： 使用工具类：
                    List<Object> list = cacheService.refreshWithPipeline(futureKey, listKey, tasks);
                    log.info("数据成功同步：{}", list);
                }
            }
        }
    }

    /**
     * 数据库数据每五分钟自动将 小于未来 五分钟的数据同步到redis中
     * 1：清理redis中的数据： 防止任务重复
     * 2：从数据库中筛选出符合条件的延时任务数据  ： 时间小于未来五分钟
     * 3：
     */
    @PostConstruct //开机自启动，每次项目启动。这个方法会启动一次
    @Scheduled(cron = "0 */5 * * * ? ")//定时任务：每分钟执行一次
    public void reloadData() {
        log.info("开始数据库数据同步Redis中...");

        //1：清理Redis中数据：防止延时任务重复执行：
        //1.1：获取Key
        Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");
        Set<String> topicKeys = cacheService.scan(ScheduleConstants.TOPIC + "*");
        //1.2:根据Key删除数据
        cacheService.delete(futureKeys);
        cacheService.delete(topicKeys);

        log.info("Redis中数据已清理...");

        //2:从数据库中筛选数据：执行时间小于未来五分钟
        //2.1：获取未来五分钟的 毫秒值
        long futureTime = System.currentTimeMillis() + (1000 * 60 * 5);
        //2.2:查询数据库找到符合条件的数据
        List<Taskinfo> taskinfos = taskinfoMapper.selectList(Wrappers.<Taskinfo>lambdaQuery()
                .le(Taskinfo::getExecuteTime, new Date(futureTime)));

        if (taskinfos != null && taskinfos.size() > 0) {
            //3 ：将数据 同步到Redis 中
            for (Taskinfo taskinfo : taskinfos) {
                TaskDto taskDto = new TaskDto();
                BeanUtils.copyProperties(taskinfo, taskDto);
                taskDto.setExecuteTime(taskinfo.getExecuteTime().getTime());
                addTaskToRedis(taskDto);
                log.info("数据库数据同步Redis中成功...");
            }
        }
    }
}

